package cn.monkey.io.transport.netty;

import cn.monkey.io.transport.core.Exchange;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DefaultNettyExchangeFactory implements NettyExchangeFactory<NettyExchange> {

    private static final Logger log = LoggerFactory.getLogger(DefaultNettyExchangeFactory.class);

    @Override
    public NettyExchange apply(String key, Exchange exchange, Channel channel) {
        if (exchange == null) {
            return new DefaultChannelExchange(key, channel).bind();
        }
        if (exchange instanceof ChannelGroupExchange nettyChannelGroupExchange) {
            ChannelGroup channelGroup = nettyChannelGroupExchange.channelGroup();
            channelGroup.removeIf(c -> !c.isActive());
            if (channelGroup.isEmpty()) {
                return new DefaultChannelExchange(key, channel).bind();
            }
            channelGroup.add(channel);
            return new DefaultChannelGroupExchange(key, channelGroup).bind();
        }
        if (exchange instanceof ChannelExchange channelExchange) {
            ChannelGroup channelGroup = new DefaultChannelGroup(channel.eventLoop(), true);
            channelGroup.add(channelExchange.channel());
            channelGroup.add(channel);
            return new DefaultChannelGroupExchange(key, channelGroup).bind();
        }
        log.error("invalid exchange type: {}", exchange.getClass());
        throw new IllegalArgumentException("invalid exchange type: " + exchange.getClass());
    }
}
